// main.go
package main

import (
	"context"
	"log"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/segmentio/kafka-go"
)

const (
	topic   = "test-topic"
	groupID = "test-group"
)

func main() {
	// 启动 pprof
	go func() {
		log.Println("pprof listening on :6060")
		if err := http.ListenAndServe(":6060", nil); err != nil {
			log.Fatalf("pprof failed: %v", err)
		}
	}()

	instanceID := os.Getenv("INSTANCE_ID")
	if instanceID == "" {
		instanceID = "default"
	}

	// 配置静态 GroupInstanceID（用于避免再平衡）
	var groupInstanceID *string
	if staticID := os.Getenv("STATIC_ID"); staticID != "" {
		groupInstanceID = &staticID
		log.Printf("[%s] Using static GroupInstanceID: %s", instanceID, staticID)
	}

	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:           []string{"localhost:9092"},
		Topic:             topic,
		GroupID:           groupID,
		GroupInstanceID:   groupInstanceID, // ✅ 正确设置方式
		SessionTimeout:    6 * time.Second,
		HeartbeatInterval: 2 * time.Second,
		MaxWait:           1 * time.Second,
		Logger:            kafka.LoggerFunc(log.Printf),
	})

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		<-sigCh
		log.Printf("[%s] Received signal, closing reader...", instanceID)
		reader.Close()
		cancel()
	}()

	log.Printf("[%s] Starting consumer...", instanceID)

	msgCount := 0
	for {
		_, err := reader.ReadMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				log.Printf("[%s] Context canceled, exiting", instanceID)
				return
			}
			log.Printf("[%s] Read error: %v", instanceID, err)
			time.Sleep(100 * time.Millisecond)
			continue
		}

		msgCount++
		if msgCount%1000 == 0 {
			log.Printf("[%s] Processed %d messages", instanceID, msgCount)
		}

		time.Sleep(10 * time.Millisecond)

		if msgCount >= 5000 {
			log.Printf("[%s] Reached 5000 messages, exiting to simulate rolling update", instanceID)
			break
		}
	}

	log.Printf("[%s] Consumer exiting", instanceID)
}
